package com.zx.learn.flink.apps.app1mysqltomysql;

import lombok.extern.slf4j.Slf4j;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;

import java.sql.*;

/**
 * MySQL 到MySQL
 */
@Slf4j
public class MySQLToMySQLMainSource {
    public static void main(String[] args) throws Exception {
        //1.env
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        //2.source
        DataStreamSource<SampleUser> sourceDS = env.addSource(new RichSourceFunction<SampleUser>() {
                                                                  PreparedStatement ps;
                                                                  Connection connection;
                                                                  Connection getConnection() {
                                                                      Connection con = null;
                                                                      try {
                                                                          Class.forName("com.mysql.cj.jdbc.Driver");
                                                                          //注意，替换成自己本地的 mysql 数据库地址和用户名、密码
                                                                          con = DriverManager.getConnection("jdbc:mysql://localhost:3306/bigdata?characterEncoding=utf8&useSSL=false&serverTimezone=UTC&rewriteBatchedStatements=true&autoReconnect=true", "root", "root");
                                                                      } catch (Exception e) {
                                                                          log.error("get connection has exception , msg = {}", e);
                                                                      }
                                                                      return con;
                                                                  }

                                                                  @Override
                                                                  public void open(Configuration parameters) throws Exception {
                                                                      super.open(parameters);
                                                                      connection = getConnection();
                                                                      String sql = "select * from sample_user limit 100";
                                                                      if (connection != null) {
                                                                          ps = this.connection.prepareStatement(sql);
                                                                      }
                                                                      log.debug("获取jdbc连接:{},{}", connection, ps);
                                                                  }

                                                                  @Override
                                                                  public void run(SourceContext<SampleUser> sourceContext) throws Exception {
                                                                      ResultSet resultSet = ps.executeQuery();
                                                                      while (resultSet.next()) {
                                                                          SampleUser obj = new SampleUser();
                                                                          obj.setId(resultSet.getInt("id"));
                                                                          obj.setName(resultSet.getString("name"));
                                                                          obj.setAge(resultSet.getInt("age"));
                                                                          sourceContext.collect(obj);
                                                                      }
                                                                  }

                                                                  @Override
                                                                  public void cancel() {

                                                                  }

                                                                  @Override
                                                                  public void close() throws Exception {
                                                                      super.close();
                                                                      //关闭连接和释放资源
                                                                      if (connection != null) {
                                                                          connection.close();
                                                                      }
                                                                      if (ps != null) {
                                                                          ps.close();
                                                                      }
                                                                  }
                                                              }
        );
        sourceDS.print();
        env.execute();
    }


}
